//! MySQL数据源实现
//! 
//! 基于sqlx的MySQL数据源插件

use async_trait::async_trait;
use sqlx::{MySql, Pool, Row, MySqlPool, Column, TypeInfo};
use serde::{Serialize, Deserialize};
use serde_json::Value;
use std::collections::HashMap;
use std::time::Instant;

use crate::{DataSource, DataSourceQuery, DataSourceResult, DataSourceStats};

/// MySQL数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MySqlConfig {
    /// 数据库连接URL
    pub database_url: String,
    /// 最大连接数
    pub max_connections: u32,
    /// 连接超时时间（秒）
    pub connect_timeout: u64,
    /// 查询超时时间（秒）
    pub query_timeout: u64,
}

impl Default for MySqlConfig {
    fn default() -> Self {
        Self {
            database_url: "mysql://root:password@localhost:3306/test".to_string(),
            max_connections: 10,
            connect_timeout: 30,
            query_timeout: 60,
        }
    }
}

/// MySQL数据源
pub struct MySqlDataSource {
    config: MySqlConfig,
    pool: Option<Pool<MySql>>,
}

impl MySqlDataSource {
    /// 创建新的MySQL数据源
    pub fn new(config: MySqlConfig) -> Self {
        Self {
            config,
            pool: None,
        }
    }
    
    /// 获取连接池
    fn get_pool(&self) -> anyhow::Result<&Pool<MySql>> {
        self.pool.as_ref().ok_or_else(|| {
            anyhow::anyhow!("MySQL数据源未连接")
        })
    }
    
    /// 将MySQL行转换为JSON对象
    fn row_to_json(row: &sqlx::mysql::MySqlRow) -> anyhow::Result<Value> {
        let mut json_obj = serde_json::Map::new();
        
        for (i, column) in row.columns().iter().enumerate() {
            let column_name = column.name();
            
            // 根据列类型转换值
            let value = match column.type_info().name() {
                "TINYINT" | "SMALLINT" | "MEDIUMINT" | "INT" | "INTEGER" => {
                    match row.try_get::<Option<i32>, _>(i) {
                        Ok(Some(v)) => Value::Number(v.into()),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
                "BIGINT" => {
                    match row.try_get::<Option<i64>, _>(i) {
                        Ok(Some(v)) => Value::Number(v.into()),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
                "FLOAT" | "DOUBLE" | "DECIMAL" => {
                    match row.try_get::<Option<f64>, _>(i) {
                        Ok(Some(v)) => Value::Number(serde_json::Number::from_f64(v).unwrap_or_else(|| 0.into())),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
                "BOOLEAN" | "BOOL" => {
                    match row.try_get::<Option<bool>, _>(i) {
                        Ok(Some(v)) => Value::Bool(v),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
                "VARCHAR" | "CHAR" | "TEXT" | "LONGTEXT" | "MEDIUMTEXT" | "TINYTEXT" => {
                    match row.try_get::<Option<String>, _>(i) {
                        Ok(Some(v)) => Value::String(v),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
                "DATE" | "DATETIME" | "TIMESTAMP" | "TIME" => {
                    match row.try_get::<Option<String>, _>(i) {
                        Ok(Some(v)) => Value::String(v),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
                _ => {
                    // 对于未知类型，尝试作为字符串获取
                    match row.try_get::<Option<String>, _>(i) {
                        Ok(Some(v)) => Value::String(v),
                        Ok(None) => Value::Null,
                        Err(_) => Value::Null,
                    }
                }
            };
            
            json_obj.insert(column_name.to_string(), value);
        }
        
        Ok(Value::Object(json_obj))
    }
}

#[async_trait]
impl DataSource for MySqlDataSource {
    fn name(&self) -> &str {
        "MySQL"
    }
    
    fn description(&self) -> &str {
        "MySQL关系型数据库数据源"
    }
    
    async fn connect(&mut self) -> anyhow::Result<()> {
        let pool = MySqlPool::connect(&self.config.database_url).await
            .map_err(|e| anyhow::anyhow!("连接MySQL失败: {}", e))?;
        
        self.pool = Some(pool);
        Ok(())
    }
    
    async fn disconnect(&mut self) -> anyhow::Result<()> {
        if let Some(pool) = &self.pool {
            pool.close().await;
        }
        self.pool = None;
        Ok(())
    }
    
    async fn is_connected(&self) -> bool {
        self.pool.is_some()
    }
    
    async fn execute_query(&mut self, query: DataSourceQuery) -> anyhow::Result<DataSourceResult> {
        let start_time = std::time::Instant::now();
        let pool = self.get_pool()?;
        
        // 执行查询
        match sqlx::query(&query.content).fetch_all(pool).await {
            Ok(rows) => {
                let duration = start_time.elapsed();
                
                // 转换结果为JSON
                let mut results = Vec::new();
                for row in &rows {
                    match Self::row_to_json(row) {
                        Ok(json_row) => results.push(json_row),
                        Err(e) => {
                            return Ok(DataSourceResult {
                                success: false,
                                data: Value::Null,
                                error: Some(format!("行数据转换失败: {}", e)),
                                stats: DataSourceStats {
                                    execution_time_ms: duration.as_millis() as f64,
                                    rows_processed: 0,
                                    bytes_processed: 0,
                                },
                            });
                        }
                    }
                }
                
                Ok(DataSourceResult {
                    success: true,
                    data: Value::Array(results),
                    error: None,
                    stats: DataSourceStats {
                        execution_time_ms: duration.as_millis() as f64,
                        rows_processed: rows.len() as u64,
                        bytes_processed: 0, // TODO: 计算实际字节数
                    },
                })
            }
            Err(e) => {
                let duration = start_time.elapsed();
                
                Ok(DataSourceResult {
                    success: false,
                    data: Value::Null,
                    error: Some(e.to_string()),
                    stats: DataSourceStats {
                        execution_time_ms: duration.as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                })
            }
        }
    }
    
    async fn health_check(&self) -> anyhow::Result<Value> {
        if !self.is_connected().await {
            return Ok(serde_json::json!({
                "status": "unhealthy",
                "database": "MySQL",
                "connected": false,
                "error": "未连接"
            }));
        }
        
        let pool = self.get_pool()?;
        
        // 执行简单的健康检查查询
        match sqlx::query("SELECT 1 as health_check").fetch_one(pool).await {
            Ok(_) => Ok(serde_json::json!({
                "status": "healthy",
                "database": "MySQL",
                "connected": true,
                "pool_size": pool.size(),
                "idle_connections": pool.num_idle()
            })),
            Err(e) => Ok(serde_json::json!({
                "status": "unhealthy",
                "database": "MySQL",
                "connected": true,
                "error": e.to_string()
            })),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_mysql_data_source_creation() {
        let config = MySqlConfig::default();
        let data_source = MySqlDataSource::new(config);
        
        assert_eq!(data_source.name(), "MySQL");
        assert_eq!(data_source.description(), "MySQL关系型数据库数据源");
        assert!(!data_source.is_connected().await);
    }

    #[tokio::test]
    async fn test_mysql_config_default() {
        let config = MySqlConfig::default();
        assert_eq!(config.database_url, "mysql://root:password@localhost:3306/test");
        assert_eq!(config.max_connections, 10);
        assert_eq!(config.connect_timeout, 30);
        assert_eq!(config.query_timeout, 60);
    }

    #[tokio::test]
    async fn test_mysql_health_check_disconnected() {
        let config = MySqlConfig::default();
        let data_source = MySqlDataSource::new(config);
        
        let health = data_source.health_check().await.unwrap();
        assert_eq!(health["status"], "unhealthy");
        assert_eq!(health["connected"], false);
    }

    // 注意：以下测试需要真实的MySQL数据库连接
    // 在CI/CD环境中可能需要跳过或使用测试数据库
    
    #[tokio::test]
    #[ignore] // 需要真实MySQL连接
    async fn test_mysql_connection() {
        let config = MySqlConfig::default();
        let mut data_source = MySqlDataSource::new(config);
        
        // 这个测试需要真实的MySQL数据库
        // 在实际环境中需要配置正确的连接字符串
        match data_source.connect().await {
            Ok(_) => {
                assert!(data_source.is_connected().await);
                data_source.disconnect().await.unwrap();
                assert!(!data_source.is_connected().await);
            }
            Err(_) => {
                // 如果没有MySQL数据库，测试会失败，这是预期的
                println!("MySQL数据库不可用，跳过连接测试");
            }
        }
    }
}
